/**
 * Copyright 2007-2008. Chongqing First Information & Network Co., Ltd. All
 * rights reserved. <a>http://www.cqfirst.com.cn</a>
 */
package com.cqcis.uip.base.framework.base;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.mina.common.ExecutorThreadModel;
import org.apache.mina.common.IoFilter;
import org.apache.mina.common.IoHandler;
import org.apache.mina.common.IoService;
import org.apache.mina.common.IoServiceConfig;
import org.apache.mina.common.IoServiceListener;
import org.apache.mina.common.IoSession;
import org.apache.mina.transport.socket.nio.SocketAcceptor;
import org.apache.mina.transport.socket.nio.SocketAcceptorConfig;
import org.apache.mina.transport.socket.nio.SocketSessionImpl;

import com.cqcis.uip.base.framework.FrameworkConstants;
import com.cqcis.uip.base.framework.UIPServiceException;
import com.cqcis.uip.base.framework.base.adaptor.AbstractSocketService;
import com.cqcis.uip.base.framework.common.Role;
import com.cqcis.uip.base.framework.common.Task;
import com.cqcis.uip.base.framework.core.RegisteredServiceContext;
import com.cqcis.uip.base.framework.filters.SocketFilterChainAdaptor;
import com.cqcis.uip.base.framework.handlers.common.AbstractServerHandler;
import com.cqcis.uip.base.framework.utils.LogServiceThreadPoolExecutor;
import com.cqcis.uip.base.framework.utils.TaskUtil;
import com.cqcis.uip.base.framework.utils.LogServiceThreadPoolExecutor.LogThreadPoolExecutor;

/**
 * Socket 服务端实现
 * 
 * @author huangmy
 * @date   2007-12-17
 */
public abstract class AbstractAcceptorSupport extends AbstractSocketService {
	private static Log log = LogFactory.getLog(AbstractAcceptorSupport.class);	
	private SocketAcceptor acceptor;        /*接口服务*/	
	private AbstractServerHandler handler;  /*业务处理*/
	private long idleTimeout;   /*socket没有数据读取空闲超时时间, 超过该定义的时间没有数据可读, 为空闲超时*/
	private SocketAcceptorConfig config = new SocketAcceptorConfig();
	private SocketAddress serviceAddress; // 服务端监听地址
	private LogThreadPoolExecutor acceptorTPE; // 监听线程池
	
	@Override
	public boolean processResult(final Task result) {		
		getExtendProcessResultThreadPoolExecutor().execute(new Runnable() {
			public void run() {
				try {					
				    handler.processResult(result);	
				} finally {
				    afterProcessResult(result);
				}
			}
		});
		
		return true;
	}
		
	public Role getRole() {		
		return Role.SOCKET_SERVER;
	}
	
	public void catchException(Task task, Throwable t) {
		try {
			handler.exceptionCaught(TaskUtil.getCurrentSession(task), t);
		} catch (Throwable e) {
			log.error("Cach Exception When catchException.", e);
		}
	}
	
	@Override
	public void start(RegisteredServiceContext serviceContext) throws Exception {		
		super.start(serviceContext);		
		int coreSize = threadPoolCoreSize/8;
		coreSize = coreSize > 2 ? coreSize : 2;
		coreSize = coreSize > 8 ? 8 : coreSize;
		acceptorTPE = (LogThreadPoolExecutor)LogServiceThreadPoolExecutor.getLogExecutor(coreSize, coreSize * 4, 60, serviceContext.getLogService(), 
				getLogConfiguration(), "(S)"+getInterfaceName());
		
		int processors = coreSize - 2;
		processors = processors > 1 ? processors : 1;
		acceptor = new SocketAcceptor(processors, acceptorTPE);
		
		config.setBacklog(30);
		config.setReuseAddress(true);
		config.setDisconnectOnUnbind(true);
        
		doStart();// 定制启动
		
		handler.registerClientNumberMax(getClientNumberMax());
		handler.registerTimeout(idleTimeout);
		handler.registerInterfaceId(getInterfaceId());
		handler.registerInterfaceLogService(getInterfaceLogService());
		handler.registerKeepAlive(isKeepAlive());
		handler.registerAdaptorRS(this);
		handler.registerTaskWaitTimeConsumePool(getTaskWaitTimeConsumePool());
		handler.registerStreamRecord(isStreamRecord());
		
		handler.start(serviceContext);
				
		try {
			serviceAddress = new InetSocketAddress(getPort());
			acceptor.bind(serviceAddress, handler, config);
		} catch (IOException e) {			
			throw e;
		}
		
		log.info("Socket Server " + getInterfaceName() + " listening on " + getPort() + " ......");
	}
	
	@Override
	public void stop(RegisteredServiceContext serviceContext) throws Exception {
		super.stop(serviceContext);
		
		handler.stop(serviceContext);
					
		final CountDownLatch latch = new CountDownLatch(1);
		if (acceptor != null) {
			acceptor.addListener(new IoServiceListener() {
				public void serviceActivated(IoService s, SocketAddress a, IoHandler h, IoServiceConfig c) {
					// do nothing
				}
				
				public void serviceDeactivated(IoService s, SocketAddress a, IoHandler h, IoServiceConfig c) {
					// 停止监听
					latch.countDown();
				}
				
				public void sessionCreated(IoSession s) {
					// do nothing					
				}
				
				public void sessionDestroyed(IoSession s) {
					// do nothing
				}
			});
			if (serviceAddress != null) {
				acceptor.unbind(serviceAddress);
			}
			
			acceptor.unbindAll(); // 停止socket监听
		}
		
		try {
		    latch.await(FrameworkConstants.RS_STOP_AWAIT_TIME, TimeUnit.MICROSECONDS);
		} catch (InterruptedException e) {//忽略异常	            
        }
		
		// 因为MINA实现，停止socket监听，采用的是异步模式（停止操作不在当前stop线程中执行），
		// 停止线程为acceptorTPE线程池中一员, 因而acceptorTPE线程池不能马上执行shutdown方法。		
		if (acceptorTPE != null) {
			// 关闭还未关闭的session
			Collection<IoSession> sessions = getManagedSessions();
			if (sessions != null) {
				for (IoSession s : sessions) {
					s.close().join();
					try {
	        		    ((SocketSessionImpl)s).getChannel().close();
	        		} catch (Throwable t) {        			
	        		}
				}
			}
			
			sessions = getManagedSessions();
			if (sessions != null) {
				for (IoSession s : sessions) {
					s.close().join();
					try {
	        		    ((SocketSessionImpl)s).getChannel().close();
	        		} catch (Throwable t) {        			
	        		}
				}
			}

			acceptorTPE.shutdownWait();
		}
		
		doStop(); // 定制停止
	}
	
	@Override
	protected boolean doSchedule() {		
		return false;
	}
	
	@Override
	public Map<String, IoFilter> getFilters() {
		final LinkedHashMap<String, IoFilter> filters = new LinkedHashMap<String, IoFilter>();
		SocketFilterChainAdaptor sofilters = new SocketFilterChainAdaptor() {			
			@Override
			public void addFirst(String name, IoFilter filter) {
				filters.put(name, filter);
				super.addFirst(name, filter);
			}
			
			@Override
			public void addLast(String name, IoFilter filter) {
				filters.put(name, filter);
				super.addLast(name, filter);
			}
		};
		
		try {
			config.getThreadModel().buildFilterChain(sofilters);
			this.acceptor.getFilterChainBuilder().buildFilterChain(sofilters);
			config.getFilterChainBuilder().buildFilterChain(sofilters);	        
		} catch (Exception e) {
			throw new UIPServiceException(e);
		}
        
		return filters;
	}
	
	@Override
	protected boolean doScheduleSync(Task task) {		
		return false;
	}
		
	protected void addFilter(String filterName, IoFilter filter) {
		if (filter != null) {
			config.getFilterChain().addLast(filterName, filter);	
		}
	}
	
	@Override
	protected void setThreadPoolExecutorModelConfig(ExecutorThreadModel etm) {
		config.setThreadModel(etm);
	}
	
	@Override
	public Collection<IoSession> getManagedSessions() {
		Collection<IoSession> sessions = new ArrayList<IoSession>();
		if (acceptor != null) {
			Set<SocketAddress> mads = acceptor.getManagedServiceAddresses();
			if (mads != null) {				
				for (SocketAddress mad : mads) {
					Collection<IoSession> ss = acceptor.getManagedSessions(mad);
					if (ss != null) {
						sessions.addAll(ss);
					}
				}
			}
		}
		return sessions;
	}
		
	/**
	 * 定制启动
	 * @throws Exception
	 */
	protected abstract void doStart() throws Exception;
	
	/**
	 * 定制停止
	 * @throws Exception
	 */
	protected abstract void doStop() throws Exception;
	
	public void setHandler(AbstractServerHandler handler) {
		this.handler = handler;
	}

	public void setIdleTimeout(long idleTimeout) {
		this.idleTimeout = idleTimeout;
	}	
}
